package com.ds.mq.consumer.converter;

import cn.hutool.json.JSONUtil;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.converter.BatchMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.HashMap;
import java.util.List;

/**
 * @class: CustomConverter
 * @description:
 * @author: fs
 * @create: 2024/8/16 11:02
 **/
public class CustomConverter implements BatchMessageConverter {
    @Override
    public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, Acknowledgment acknowledgment, Consumer<?, ?> consumer, Type payloadType) {
        Headers headers = records.get(0).headers();
        HashMap<String, Object> hashMap = new HashMap<>();
        for (Header header : headers) {
            hashMap.put(header.key(), header.value());
        }
        ParameterizedType genericSuperclass = (ParameterizedType) payloadType;
        Class clazz = (Class<?>) genericSuperclass.getActualTypeArguments()[0];
        List<?> events = records.stream().map(record -> JSONUtil.toBean(record.value().toString(),clazz)).toList();
        return MessageBuilder.createMessage(events, new MessageHeaders(hashMap));
    }

    @Override
    public List<ProducerRecord<?, ?>> fromMessage(Message<?> message, String defaultTopic) {
        return List.of();
    }
}